package com.quanyan.club.mq.consumer;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.google.common.base.Strings;
import com.quanyan.club.entity.RespClubXpLevelMsg;
import com.quanyan.club.entity.db.TbClubExperienceHistory;
import com.quanyan.club.entity.db.TbClubMsgConsumerRecord;
import com.quanyan.club.entity.db.TbLevel;
import com.quanyan.club.entity.vo.resp.RespClubExperienceHistory;
import com.quanyan.club.entity.vo.resp.RespExperienceParameter;
import com.quanyan.club.mapper.TbClubMsgConsumerRecordMapper;
import com.quanyan.club.mapper.TbLevelMapper;
import com.quanyan.club.mq.constant.MqConstant;
import com.quanyan.club.service.ClubService;
import com.quanyan.club.service.ClubXpLevelService;
import com.quanyan.common.constants.Constants;
import com.quanyan.common.constants.MqConstants;
import com.quanyan.common.utils.ObjectUtil;
import com.quanyan.mq.enums.MqConsumerStatus;
import com.quanyan.mq.mq.consumer.AbstractMqConsumer;
import com.quanyan.mq.mq.consumer.MqTopic;
import com.quanyan.orderpay.service.OrderPayService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.List;

/**
 * Created by zhang on 16/7/22.
 */
@Component("clubXpLevelConsumer")
public class ClubXpLevelConsumer extends AbstractMqConsumer{

    private final static Logger logger = LoggerFactory.getLogger(ClubXpLevelConsumer.class);

    private DefaultMQPushConsumer consumer;

    private MqTopic[] mqTopics;

    @Autowired
    private ClubXpLevelService clubXpLevelService;

    @Autowired
    private TbClubMsgConsumerRecordMapper clubMsgConsumerRecordMapper;

    @Value("${mq.nameServer.address}")
    private String nameServerAddress;

    @Autowired
    private TbLevelMapper levelMapper;

    @Override
    public void initConsumer() throws MQClientException {
        consumer = new DefaultMQPushConsumer(getConsumerGroup());//保证应用内唯一
        consumer.setNamesrvAddr(getNameServerAddress());
        consumer.setInstanceName(getInstanceName());//保证应用内唯一
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //设置消费间隔
        consumer.setPullInterval(200);
        consumer.setPullBatchSize(1);
        mqTopics = getMqTopics();
        if (mqTopics != null && mqTopics.length > 0) {
            for (MqTopic mqTopic : mqTopics) {
                consumer.subscribe(mqTopic.getTopicName(),
                        Strings.isNullOrEmpty(mqTopic.getTags()) ? "*" : mqTopic.getTags());
            }
        }
    }

    public void start() {
        try {
            initConsumer();
            consumer.registerMessageListener(new MessageListenerConcurrently() {

                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages,
                                                                ConsumeConcurrentlyContext context) {
                    //消费前根据全局唯一的msgId在Db中判断是否消费

                    /**
                     * 默认messages里只有一条消息，可以设置consumeMessageBatchMaxSize参数来批量接收消息
                     * consumer.setConsumeMessageBatchMaxSize(8);
                     **/

                    logger.warn(Thread.currentThread().getName() + " receive messages: "
                            + messages.toString() + "size: " + messages.size());

                    MqConsumerStatus mqConsumerStatus = process(messages, context);

                    if (mqConsumerStatus.equals(MqConsumerStatus.RE_CONSUME_LATER)) {
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            consumer.start();

        } catch (MQClientException e) {
            e.printStackTrace();
        }

        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            public void run() {
                consumer.shutdown();
                logger.info("consumer调用addShutdownHook方法优雅关闭 ！！！");
            }
        }));
    }

    @PostConstruct
    public void run() {
        logger.warn("nameServerAddress : " + nameServerAddress + "--" +System.currentTimeMillis());
        this.start();
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, isolation = Isolation.REPEATABLE_READ, rollbackFor = Exception.class)
    protected MqConsumerStatus process(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
        logger.info("俱乐部等级经验Consumer开始工作");
        MessageExt messageExt = messages.get(0);
        //TbClubMsgConsumerRecord clubMsgConsumerRecord = buildMsgConsumerRecord(messageExt);
        if(!MqConstants.CLUB_XP_LEVEL_TOPIC_NAME.equals(messageExt.getTopic())){
            //向消息记录表中添加数据
            //clubMsgConsumerRecord.setStatus(false);
            //clubMsgConsumerRecordMapper.insertSelective(clubMsgConsumerRecord);
            return MqConsumerStatus.RE_CONSUME_LATER;
        }
        String msgId = messageExt.getKeys();

        byte[] bodyByte = messageExt.getBody();
        RespClubXpLevelMsg respClubXpLevelMsg = (RespClubXpLevelMsg) ObjectUtil.toObject(bodyByte);
        int uid = respClubXpLevelMsg.getUid();
        int clubId = respClubXpLevelMsg.getClubId();
        String expName = respClubXpLevelMsg.getExpName();
        try {
            RespExperienceParameter respExperienceParameter = clubXpLevelService.getExperienceParameter(expName);
            if (respExperienceParameter.getId() == null){
                //向消息记录表中添加数据
                //clubMsgConsumerRecord.setStatus(false);
                //clubMsgConsumerRecordMapper.insertSelective(clubMsgConsumerRecord);
                return MqConsumerStatus.RE_CONSUME_LATER;
            }

            TbClubExperienceHistory clubExpHistory = getClubExperienceHistory(uid, clubId, expName, respExperienceParameter);

            if(respExperienceParameter.getType() != null && respExperienceParameter.getType().intValue() == 0){ //0表示一次性经验
                //1 查找经验历史表中是否有此条记录
                boolean isExist = clubXpLevelService.isExistClubExperienceHistory(expName,clubId);
                if (!isExist){
                    clubExpHistory.setIsEnable((byte)1); //此记录不可用
                    //2 向经验历史表中插入数据
                    clubXpLevelService.insertClubExperienceHistory(clubExpHistory);

                    //3 更新俱乐部等级
                    clubXpLevelService.modifyClubExperience(clubId,respExperienceParameter.getExperience());

                }
            }else if (respExperienceParameter.getType() != null && respExperienceParameter.getType().intValue() == 1){ //1 表示每日记录
                Integer count = clubXpLevelService.countByCurDate(expName,clubId);
                if (count == null)
                    count = 0;
                if(count.intValue() == 0 && respExperienceParameter.getExpNumber() != null){
                    if(respExperienceParameter.getExpNumber()>1){
                        clubExpHistory.setIsEnable((byte)0); //可用
                    }else {
                        clubExpHistory.setIsEnable((byte)1); //不可用
                    }
                    // 向经验历史表中插入数据
                    clubXpLevelService.insertClubExperienceHistory(clubExpHistory);
                   // 更新俱乐部等级
                    clubXpLevelService.modifyClubExperience(clubId,respExperienceParameter.getExperience());

                }else {
                    if (respExperienceParameter.getExpNumber() != null && count < respExperienceParameter.getExpNumber() ){  //可用
                        //插入经验历史表
                        if (respExperienceParameter.getExpNumber() == (count+1)){
                            clubExpHistory.setIsEnable((byte)1); //记录不可用
                        }else {
                            clubExpHistory.setIsEnable((byte)0); //记录可用
                        }
                        clubXpLevelService.insertClubExperienceHistory(clubExpHistory);
                        //更新俱乐部等级
                        clubXpLevelService.modifyClubExperience(clubId,respExperienceParameter.getExperience());

                    }
                }
            }
            //向消息记录表中添加数据
            //clubMsgConsumerRecord.setStatus(true);
            //clubMsgConsumerRecordMapper.insertSelective(clubMsgConsumerRecord);
            return MqConsumerStatus.CONSUMER_SUCCESS;
        }catch(Exception e){
            logger.error("消费异常，msgId【{}】，topic【{}】,tag【{}】,errorMsg【{}】", msgId,
                    OrderPayService.TOPIC_ORDER_STATUS_NOTIFY,
                    Constants.CLUB_SYSTEM_CODE, e.getMessage());
            //消费异常,记录msgId 到 db 中
            //clubMsgConsumerRecord.setStatus(false);
            //clubMsgConsumerRecordMapper.insertSelective(clubMsgConsumerRecord);
            return MqConsumerStatus.RE_CONSUME_LATER;
        }
    }

    private TbClubMsgConsumerRecord buildMsgConsumerRecord(MessageExt messageExt) {
        TbClubMsgConsumerRecord clubMsgConsumerRecord = new TbClubMsgConsumerRecord();
        clubMsgConsumerRecord.setTopic(messageExt.getTopic());
        clubMsgConsumerRecord.setMsgId(messageExt.getMsgId());
        clubMsgConsumerRecord.setTag(messageExt.getTags());
        return clubMsgConsumerRecord;
    }

    private TbClubExperienceHistory getClubExperienceHistory(int uid, int clubId, String expName, RespExperienceParameter respExperienceParameter) {
        TbClubExperienceHistory clubExpHistory = new TbClubExperienceHistory();
        if (respExperienceParameter.getId() != null){
            clubExpHistory.setClubId(clubId);
            clubExpHistory.setExpName(expName);
            clubExpHistory.setExperience(respExperienceParameter.getExperience());
            clubExpHistory.setIsDelete((byte)0);
            clubExpHistory.setUid(uid);
        }
        return clubExpHistory;
    }

    @Override
    protected String getConsumerGroup() {
        return MqConstant.CLUB_XP_LEVEL_GROUP_NAME;
    }

    @Override
    protected String getNameServerAddress() {
        return this.nameServerAddress;
    }

    @Override
    protected String getInstanceName() {
        return MqConstant.CLUB_XP_LEVEL_INSTANCE_NAME;
    }

    @Override
    protected MqTopic[] getMqTopics() {
        MqTopic[] mqTopics = new MqTopic[1];
        mqTopics[0] = new MqTopic(MqConstants.CLUB_XP_LEVEL_TOPIC_NAME ,MqConstants.CLUB_XP_LEVEL_TAGS_NAME);
        return mqTopics;
    }
}
